//package com.hb.resource.consumer;
//
//import com.google.common.collect.Lists;
//import org.apache.kafka.clients.consumer.ConsumerConfig;
//import org.apache.kafka.clients.consumer.ConsumerRecords;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//
//import java.time.Duration;
//import java.time.temporal.ChronoUnit;
//import java.util.List;
//import java.util.Properties;
//
///**
// * kafka消费者 不依赖spring环境
// *
// * @author hb
// * @date:2023/4/25 14:53
// * @date 2023/04/26
// */
//
//public class KafkaCustomConsumer {
//
//    /**
//     * 主要
//     *
//     * @param args arg游戏
//     */
//    public static void main(String[] args) {
//        Properties init = init();
//        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(init);
//        normalSubscribe(kafkaConsumer, Lists.newArrayList("testTopicNew2","testTopicNew1"));
//        kafkaConsumer.close();
//    }
//
//    /**
//     * 初始化
//     *
//     * @return {@link Properties}
//     */
//    public static Properties init() {
//        Properties prop = new Properties();
//        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
//        // 自动提交的延时
//        prop.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
//        prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
//        // 反序列化
//        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//
//        prop.put(ConsumerConfig.GROUP_ID_CONFIG, "localhost_1");
//
//        // 1. 新消费者组会从头开始消费
//        // 2. 当前的offset位置的数据不存在了, 消费过的数据不存在了.
//        // 3. 如何重新消费一个topic: 换组名+earliest
//        prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");// earliest, latest
//        return prop;
//    }
//
//    /**
//     * 正常的订阅
//     *
//     * @param consumer 消费者
//     * @param topics   主题
//     */
//    public static void normalSubscribe(KafkaConsumer<String, String> consumer, List<String> topics) {
//        consumer.subscribe(topics);
//        while (true) {
//            ConsumerRecords<String, String> poll = consumer.poll(Duration.of(1, ChronoUnit.SECONDS));
//            poll.iterator().forEachRemaining(i -> {
//                System.out.println(i.key()+"---"+i.value());
//            });
////			consumer.commitSync();
//            consumer.commitAsync();
//        }
//    }
//
//
//}
